package cloud

import (
	"context"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"sync"

	"gitee.com/rw555/fdsmastergo/pkg/config"
	"gitee.com/rw555/fdsmastergo/pkg/logger"
	"go.uber.org/zap"
)

// Subscriber MQTT消息订阅器
type Subscriber struct {
	connMgr      *ConnectionManager
	configClient *config.Client
	publisher    *Publisher
	wg           sync.WaitGroup
	stopChan     chan struct{}
	msgChan      chan mqttMessage
}

// mqttMessage MQTT消息结构
type mqttMessage struct {
	topic   string
	message string
}

// NewSubscriber 创建消息订阅器
func NewSubscriber(connMgr *ConnectionManager, configClient *config.Client) *Subscriber {
	return &Subscriber{
		connMgr:      connMgr,
		configClient: configClient,
		stopChan:     make(chan struct{}),
		msgChan:      make(chan mqttMessage, 32), // 设置合适的缓冲区大小
	}
}

// NewSubscriber 创建消息订阅器
func (s *Subscriber) SetPublisher(publisher *Publisher) *Subscriber {
	s.publisher = publisher
	return s
}

// Start 启动消息订阅器
func (s *Subscriber) Start(ctx context.Context) error {
	// 订阅所有主题
	subscribeTopics := s.configClient.GetConfigsByPrefix(s.connMgr.systemConfig, "mqtt.subscribe.")
	if subscribeTopics == nil {
		return fmt.Errorf("订阅主题配置格式错误")
	}

	for _, topicInterface := range subscribeTopics {
		topic, ok := topicInterface.(string)
		if !ok {
			logger.SystemError("订阅主题格式错误", zap.Any("topic", topicInterface))
			continue
		}
		if err := s.subscribeTopic(topic); err != nil {
			return fmt.Errorf("订阅主题 %s 失败: %w", topic, err)
		}
	}

	// 启动消息处理协程
	s.wg.Add(1)
	go s.messageLoop(ctx)

	logger.SystemInfo("消息订阅器启动成功")
	return nil
}

// subscribeTopic 订阅单个主题
func (s *Subscriber) subscribeTopic(topic string) error {
	// 替换主题中的设备号
	deviceNumber := s.connMgr.systemConfig["mqtt.device_number"].(string)
	if deviceNumber == "" {
		return fmt.Errorf("设备号配置格式错误")
	}

	actualTopic := strings.ReplaceAll(topic, "{device_number}", deviceNumber)
	handler := func(msg string) {
		select {
		case s.msgChan <- mqttMessage{topic: actualTopic, message: msg}:
			// 消息已发送到通道
		default:
			logger.SystemError("消息通道已满，丢弃消息",
				zap.String("topic", actualTopic),
				zap.String("message", msg))
		}
	}

	if err := s.connMgr.SubscribeMQTT(actualTopic, handler); err != nil {
		return fmt.Errorf("MQTT订阅失败: %w", err)
	}

	logger.SystemInfo("订阅主题成功",
		zap.String("actual_topic", actualTopic))
	return nil
}

// messageLoop 消息处理循环
func (s *Subscriber) messageLoop(ctx context.Context) {
	defer s.wg.Done()
	logger.SystemInfo("启动消息处理协程")

	for {
		select {
		case <-ctx.Done():
			return
		case <-s.stopChan:
			return
		case msg := <-s.msgChan:
			s.handleMessage(msg.topic, msg.message)
		}
	}
}

// Stop 停止消息订阅器
func (s *Subscriber) Stop() {
	close(s.stopChan)
	s.wg.Wait()
	close(s.msgChan)
	logger.SystemInfo("消息订阅器已停止")
}

// handleMessage 处理接收到的消息
func (s *Subscriber) handleMessage(topic, message string) {
	// 解析消息
	var data map[string]interface{}
	if err := json.Unmarshal([]byte(message), &data); err != nil {
		logger.SystemError("解析消息失败",
			zap.String("topic", topic),
			zap.Error(err))
		return
	}

	// 根据主题类型处理消息，参考：cmd\config_server\default_configs.json
	if strings.Contains(topic, "devices/telemetry/control/") {
		s.handleTelemetryControl(data)
	} else if strings.Contains(topic, "devices/attributes/set/") {
		s.handleAttributesSet(data)
	} else if strings.Contains(topic, "devices/attributes/get/") {
		s.handleAttributesGet(data)
	} else if strings.Contains(topic, "devices/command/") {
		s.handleCommandSet(data)
	} else if strings.Contains(topic, "ota/devices/inform/") {
		s.handleResponse("OTA通知", data)
	} else if strings.Contains(topic, "devices/attributes/response/") {
		s.handleResponse("属性", data)
	} else if strings.Contains(topic, "devices/event/response/") {
		s.handleResponse("事件", data)
	} else {
		logger.SystemInfo("收到未知消息", zap.String("topic", topic), zap.String("message", message))
	}
}

func (s *Subscriber) handleResponse(key string, data map[string]interface{}) {
	// 检查必需字段
	result, ok := data["result"].(float64)
	if ok && result != 0 {
		logger.SystemDebug(key+"发布应答为 成功", zap.String("result", strconv.FormatFloat(result, 'f', -1, 64)))
		return
	}

	message, ok := data["message"].(string)
	if ok && strings.Contains(message, "success") {
		logger.SystemDebug(key+"发布应答为 成功", zap.String("message", message))
		return
	}

	// 获取可选字段
	errcode, _ := data["errcode"].(string)
	ts, _ := data["ts"].(float64)
	method, _ := data["method"].(string)

	logger.SystemError(key+"发布应答为 失败",
		zap.String("errcode", errcode),
		zap.Float64("ts", ts),
		zap.String("method", method))
}

func (s *Subscriber) handleTelemetryControl(data map[string]interface{}) {
	logger.SystemInfo("遥测控制", zap.Any("data", data))
}

func (s *Subscriber) handleAttributesSet(data map[string]interface{}) {
	logger.SystemInfo("属性设置", zap.Any("data", data))
	s.publisher.PublishResponse("devices/attributes/set/response/{message_id}", 0, "success")
}

func (s *Subscriber) handleAttributesGet(data map[string]interface{}) {
	logger.SystemInfo("属性获取", zap.Any("data", data))
}

func (s *Subscriber) handleCommandSet(data map[string]interface{}) {
	logger.SystemInfo("命令", zap.Any("data", data))
	s.publisher.PublishResponse("devices/command/response/{message_id}", 0, "success")
}
